package org.springblade.common.utils.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;

/**
 * @Date 14:15 2023/5/29
 * @Description {
 * canal client
 * }
 * @Author lsay
 **/
//@Component
@Slf4j
public class CanalClientUtil implements ApplicationRunner {

	@Value("${canal.client.instances.example.host}")
	private String canalHost;

	@Value("${canal.client.instances.example.port}")
	private Integer canalPort;

	@Resource
	MessageHandler messageHandler;

	private static final String DB_NAME = "sida_maintain";
	private static final String TABLE_NAME =
		"sida_maintain.blade_user," +
			"sida_maintain.blade_tenant," +
			"sida_maintain.blade_estate_elevator," +
			"sida_maintain.blade_estate_elevator_user," +
			"sida_maintain.blade_elevator," +
			"sida_maintain.blade_elevator_detail," +
			"sida_maintain.blade_building";

	@Override
	public void run(ApplicationArguments args) throws Exception {
		CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalHost, canalPort), "example", "", "");
		try {
			//打开连接
			connector.connect();
			log.info("Canal Client Service Starting Success ...");
			//订阅数据库表，全部表
			connector.subscribe(TABLE_NAME);
			//回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始
			connector.rollback();
			while (true) {
				//获取指定数量的数据
				Message message = connector.getWithoutAck(1000);
				//获取批量ID
				long batchId = message.getId();
				//获取批量的数量
				int size = message.getEntries().size();
				//如果没有数据
				if (batchId == -1 || size == 0) {
					try {
						//现成休眠2s
						Thread.sleep(2000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				} else {
					//如果有数据，处理数据
					messageHandler.handler(message);
				}
				connector.ack(batchId);
			}

		} catch (Exception e) {
			log.error("Canal Client Service Starting Error ...");
			e.printStackTrace();
		} finally {
			connector.disconnect();

		}

	}
}
